package mongodb.spark.examples;

import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import com.mongodb.spark.MongoSpark;

import mongodb.spark.examples.model.Coll;

public final class DatasetSQLDemo {

	public static void main(final String[] args) throws InterruptedException {

		String uri = "mongodb://192.168.95.128/test.characters";
		SparkSession spark = SparkSession.builder().master("local").appName("MongoSparkConnectorIntro")
				.config("spark.mongodb.input.uri", uri).config("spark.mongodb.output.uri", uri).getOrCreate();

		// Create a JavaSparkContext using the SparkSession's SparkContext object
		JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());

		System.out.println("1----------------------------------------");
		// Load data and infer schema, disregard toDF() name as it returns Dataset
		Dataset<Row> implicitDS = MongoSpark.load(jsc).toDF();
		implicitDS.printSchema();
		implicitDS.show();

		System.out.println("2----------------------------------------");
		// Load data with explicit schema
		Dataset<Coll> explicitDS = MongoSpark.load(jsc).toDS(Coll.class);
		explicitDS.printSchema();
		explicitDS.show();
		System.out.println("3----------------------------------------");

		// Create the temp view and execute the query
		explicitDS.createOrReplaceTempView("characters");
		Dataset<Row> centenarians = spark.sql("SELECT name, age FROM characters WHERE age >= 100");
		centenarians.show();
		System.out.println("4----------------------------------------");

		// Write the data to the "hundredClub" collection
		MongoSpark.write(centenarians).option("collection", "hundredClub").mode("overwrite").save();
		System.out.println("1----------------------------------------");

		// Load the data from the "hundredClub" collection
//		MongoSpark.load(sparkSession, ReadConfig.create(sparkSession).withOption("collection", "hundredClub"),
//				Character.class).show();

		jsc.close();

	}
}